/*
 * Copyright (C) 2009 Brent Fulgham.  All rights reserved.
 * Copyright (C) 2009 Google Inc.  All rights reserved.
 * Copyright (C) 2012 Sony Network Entertainment Intl.  All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are
 * met:
 *
 *     * Redistributions of source code must retain the above copyright
 * notice, this list of conditions and the following disclaimer.
 *     * Redistributions in binary form must reproduce the above
 * copyright notice, this list of conditions and the following disclaimer
 * in the documentation and/or other materials provided with the
 * distribution.
 *     * Neither the name of Google Inc. nor the names of its
 * contributors may be used to endorse or promote products derived from
 * this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

#include "config.h"
#include "SocketStreamHandle.h"

#include "Logging.h"
#include "NotImplemented.h"
#include "RunLoop.h"
#include "SocketStreamError.h"
#include "SocketStreamHandleClient.h"
#include <errno.h>
#include <manx/NetworkProfile.h>
#include <manx/Ssl.h>
#include <sys/ioctl.h> // for FIONREAD
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>
#include <wtf/Deque.h>
#include <wtf/RefPtr.h>
#include <wtf/Threading.h>
#include <wtf/Vector.h>
#include <wtf/text/CString.h>

// #define SOCKETSTREAMHANDLE_DEBUG_OUTPUT
// #define SOCKETSTREAMHANDLE_DEBUG_OUTPUT_SOCKET_DATA
#ifdef SOCKETSTREAMHANDLE_DEBUG_OUTPUT
#define DEBUG_PRINTF(...) printf(__VA_ARGS__)
#else
#define DEBUG_PRINTF(...)
#endif

namespace {

curl_socket_t curlopensocketCallback(void* clientp, curlsocktype, struct curl_sockaddr*)
{
    // Just return the external socket (passed by OPENSOCKETDATA) untouched
    return *(curl_socket_t *)clientp;
}

int curlsockoptCallback(void*, curl_socket_t, curlsocktype)
{
    return CURL_SOCKOPT_ALREADY_CONNECTED;
}

CURLcode curlsslctxCallback(CURL*, void* sslctx, void* param)
{
    Manx::Ssl::sslctxfunForWebSocket(sslctx, param);
    return CURLE_OK;
}

}

namespace WebCore {

class SocketStreamHandleManager {
    WTF_MAKE_NONCOPYABLE(SocketStreamHandleManager);
public:
    static SocketStreamHandleManager& shared();

    void initializeStream(PassRefPtr<SocketStreamHandle>);
    void terminateStream(PassRefPtr<SocketStreamHandle>);

    void enableReadyWriteNotification(PassRefPtr<SocketStreamHandle>);
    void enableReadyReadNotification(PassRefPtr<SocketStreamHandle>);

private:
    SocketStreamHandleManager();

    void setupSocketNotifierPipe();

    static void socketNotifierThread(void* context);
    void wakeUpSocketNotifierThread();

    void dispatchReadyRead(PassRefPtr<SocketStreamHandle>);
    void dispatchReadyWrite(PassRefPtr<SocketStreamHandle>);
    void dispatchRemoteDisconnected(PassRefPtr<SocketStreamHandle>);

    typedef Vector<RefPtr<SocketStreamHandle> > StreamHandleVector;
    StreamHandleVector m_streams;
    Mutex m_streamsMutex;

    ThreadIdentifier m_socketNotifierThreadId;

    int m_readPipeDescriptor;
    int m_writePipeDescriptor;
};

PassRefPtr<SocketStreamHandle> SocketStreamHandle::create(const KURL& url, SocketStreamHandleClient* client)
{
    RefPtr<SocketStreamHandle> handle = adoptRef(new SocketStreamHandle(url, client));
    if (client)
        RunLoop::current()->dispatch(bind(&SocketStreamHandle::initialize, handle.get()));

    return handle.release();
}

PassRefPtr<SocketStreamHandle> SocketStreamHandle::create(int socketfd, SocketStreamHandleClient* client)
{
    RefPtr<SocketStreamHandle> handle = adoptRef(new SocketStreamHandle(socketfd, client));
    if (client)
        RunLoop::current()->dispatch(bind(&SocketStreamHandle::initialize, handle.get()));

    return handle.release();
}


SocketStreamHandle::SocketStreamHandle(const KURL& url, SocketStreamHandleClient* client)
    : SocketStreamHandleBase(url, client)
    , m_handle(0)
    , m_socket(-1)
    , m_hasPendingDataForSending(false)
    , m_hasPendingReadOrDisconnect(false)
    , m_isDisconnected(false)
    , m_ownerRunLoop(0)
{
}

SocketStreamHandle::SocketStreamHandle(int socketfd, SocketStreamHandleClient *client)
    : SocketStreamHandleBase(KURL(), client)
    , m_handle(0)
    , m_socket(socketfd)
    , m_hasPendingDataForSending(false)
    , m_hasPendingReadOrDisconnect(false)
    , m_isDisconnected(false)
    , m_ownerRunLoop(0)
{
}

void SocketStreamHandle::initialize()
{
    if (m_state != SocketStreamHandleBase::Connecting)
        return;

    DEBUG_PRINTF("SocketStreamHandle: %p created with client %p and socket %d\n", this, m_client, m_socket);
    SocketStreamHandleManager::shared().initializeStream(this);
}

SocketStreamHandle::~SocketStreamHandle()
{
    m_socket = 0x29292929; // hexspeak
    
    m_handle = 0;
    m_hasPendingDataForSending = false;
    m_hasPendingReadOrDisconnect = false;
    m_isDisconnected = true;
    m_ownerRunLoop = 0;

    ASSERT(m_state == SocketStreamHandleBase::Closed);
    DEBUG_PRINTF("SocketStreamHandle: %p destroyed\n", this);
}

void SocketStreamHandle::readyRead()
{
    ASSERT(RunLoop::current() == m_ownerRunLoop);

    if (m_handle) {
        char buffer[1024];
        const size_t readSize = sizeof(buffer) / sizeof(*buffer);
        size_t nb;
        CURLcode res;
        if ((res = curl_easy_recv(m_handle, buffer, readSize, &nb)) != CURLE_OK) {
            LOG_ERROR("SocketStreamHandle: [%d] Receive error %d: %s", m_socket, res, curl_easy_strerror(res));
            return;
        }
        DEBUG_PRINTF("[%d] Received %4ld bytes\n", m_socket, nb);
#ifdef SOCKETSTREAMHANDLE_DEBUG_OUTPUT_SOCKET_DATA
        DEBUG_PRINTF("     Data received: %*s\n", nb, buffer);
#endif

        SocketStreamHandleManager::shared().enableReadyReadNotification(this);

        if (client())
            client()->didReceiveSocketStreamData(this, buffer, nb);
    } else {
        // This can happen if the handle was closed while we were dispatching the notification
        DEBUG_PRINTF("SocketStreamHandle: detected stale handle in readyRead()! Skipping read.\n");
    }
}

void SocketStreamHandle::readyWrite()
{
    ASSERT(RunLoop::current() == m_ownerRunLoop);

    if (m_handle)
        sendPendingData();
    else
        DEBUG_PRINTF("SocketStreamHandle: detected stale handle in readyWrite()! Skipping write.\n");
}

void SocketStreamHandle::remoteDisconnected()
{
    DEBUG_PRINTF(">>> SocketStreamHandle::%s()\n", __FUNCTION__);
    ASSERT(RunLoop::current() == m_ownerRunLoop);

    if (m_handle)
        platformClose();
    else
        DEBUG_PRINTF("SocketStreamHandle: detected stale handle in remoteDisconnected()! Skipping close.\n");
}


int SocketStreamHandle::platformSend(const char* data, int length)
{
    ASSERT(RunLoop::current() == m_ownerRunLoop);

    if (!data || length <= 0)
        return 0;

    CURLcode res;
    size_t nb = 0;

#ifdef SOCKETSTREAMHANDLE_DEBUG_OUTPUT_SOCKET_DATA
    DEBUG_PRINTF("[%d] Sending data %*s\n", m_socket, length, data);
#endif
    if ((res = curl_easy_send(m_handle, data, length, &nb)) != CURLE_OK) {
        LOG_ERROR("SocketStreamHandle: [%d] Send error %d: %s", m_socket, res, curl_easy_strerror(res));
        return -1;
    }
    DEBUG_PRINTF("[%d] Data sent (%ld/%d bytes).\n", m_socket, nb, length);

    if (nb < length)
        SocketStreamHandleManager::shared().enableReadyWriteNotification(this);

    return nb;
}

void SocketStreamHandle::platformClose()
{
    DEBUG_PRINTF(">>> SocketStreamHandle::%s()\n", __FUNCTION__);
    DEBUG_PRINTF("SocketStreamHandle: %p closed\n", this);
    ASSERT(RunLoop::current() == m_ownerRunLoop);

    SocketStreamHandleManager::shared().terminateStream(this);
    if (client())
        client()->didCloseSocketStream(this);
}

void SocketStreamHandle::didReceiveAuthenticationChallenge(const AuthenticationChallenge&)
{
    notImplemented();
}

void SocketStreamHandle::receivedCredential(const AuthenticationChallenge&, const Credential&)
{
    notImplemented();
}

void SocketStreamHandle::receivedRequestToContinueWithoutCredential(const AuthenticationChallenge&)
{
    notImplemented();
}

void SocketStreamHandle::receivedCancellation(const AuthenticationChallenge&)
{
    notImplemented();
}

SocketStreamHandleManager& SocketStreamHandleManager::shared()
{
    DEFINE_STATIC_LOCAL(SocketStreamHandleManager, socketStreamManager, ());
    return socketStreamManager;
}

void SocketStreamHandleManager::initializeStream(PassRefPtr<SocketStreamHandle> prpHandle)
{
    if (!m_socketNotifierThreadId) {
        LOG_ERROR("SocketStreamHandleManager: no socket notifier, can't add stream!");
        return;
    }

    RefPtr<SocketStreamHandle> handle = prpHandle;

    handle->m_ownerRunLoop = RunLoop::current();
    handle->m_handle = curl_easy_init();
    if (!handle->m_handle) {
        LOG_ERROR("SocketStreamHandleManager: %p couldn't create CURL handle!", handle.get());
        return;
    }

    if (handle->client())
        handle->client()->willOpenSocketStream(handle.get());

    long curlsocketfd = 0;
    CURLcode res;
    if (handle->m_socket >= 0) {
        // Set up curl to use the external socketfd and pass a bogus URL (needed but not used)
        if ((res = curl_easy_setopt(handle->m_handle, CURLOPT_URL, "http://1.2.3.4:12345")) != CURLE_OK)
            goto curlerror;
        if ((res = curl_easy_setopt(handle->m_handle, CURLOPT_OPENSOCKETFUNCTION, curlopensocketCallback)) != CURLE_OK)
            goto curlerror;
        if ((res = curl_easy_setopt(handle->m_handle, CURLOPT_OPENSOCKETDATA, &handle->m_socket)) != CURLE_OK)
            goto curlerror;
        if ((res = curl_easy_setopt(handle->m_handle, CURLOPT_SOCKOPTFUNCTION, curlsockoptCallback)) != CURLE_OK)
            goto curlerror;
    } else {
        // Curl doesn't support the ws/wss protocols, replace them with http/https
        // This is just to make the parser happy and let us connecting
        WTF::String url(handle->m_url.protocolIs("wss") ? "https://" : "http://");
        url.append(handle->m_url.string().substring(url.length()-2));

        DEBUG_PRINTF("Adding stream for URL %s\n", url.utf8().data());

        if ((res = curl_easy_setopt(handle->m_handle, CURLOPT_URL, url.utf8().data())) != CURLE_OK)
            goto curlerror;
    }

    if ((res = curl_easy_setopt(handle->m_handle, CURLOPT_SSL_VERIFYHOST, Manx::NetworkProfile::get(Manx::NetPropsSslVerifyHost))) != CURLE_OK)
        goto curlerror;
    if ((res = curl_easy_setopt(handle->m_handle, CURLOPT_SSL_VERIFYPEER, Manx::NetworkProfile::get(Manx::NetPropsSslVerifyPeer))) != CURLE_OK)
        goto curlerror;
    if ((res = curl_easy_setopt(handle->m_handle, CURLOPT_SSL_CIPHER_LIST, Manx::NetworkProfile::getString(Manx::NetPropsSslCipherList))) != CURLE_OK)
        goto curlerror;
    if (Manx::NetworkProfile::get(Manx::NetPropsSslVerifyPeer)
        && (res = curl_easy_setopt(handle->m_handle, CURLOPT_SSL_CTX_FUNCTION, curlsslctxCallback)) != CURLE_OK)
        goto curlerror;

    if ((res = curl_easy_setopt(handle->m_handle, CURLOPT_CONNECT_ONLY, 1L)) != CURLE_OK)
        goto curlerror;
    if ((res = curl_easy_perform(handle->m_handle)) != CURLE_OK)
        goto curlerror;
    if ((res = curl_easy_getinfo(handle->m_handle, CURLINFO_LASTSOCKET, &curlsocketfd)) != CURLE_OK)
        goto curlerror;

    ASSERT(handle->m_socket == -1 || curlsocketfd == handle->m_socket);

    handle->m_socket = static_cast<curl_socket_t>(curlsocketfd);
    handle->m_state = SocketStreamHandleBase::Open;
    if (handle->client())
        handle->client()->didOpenSocketStream(handle.get());

    m_streamsMutex.lock();
    m_streams.append(handle);
    m_streamsMutex.unlock();

    wakeUpSocketNotifierThread();

    return; // success!

curlerror:
    LOG_ERROR("SocketStreamHandleManager: %p curl Error %d: %s", handle.get(), res, curl_easy_strerror(res));
    curl_easy_cleanup(handle->m_handle);
    if (handle->client())
        handle->client()->didFailSocketStream(handle.get(), SocketStreamError(res));

    return;
}

void SocketStreamHandleManager::terminateStream(PassRefPtr<SocketStreamHandle> prpHandle)
{
    DEBUG_PRINTF(">>> SocketStreamHandle::%s()\n", __FUNCTION__);
    ASSERT(prpHandle);

    RefPtr<SocketStreamHandle> handle = prpHandle;

    ASSERT(RunLoop::current() == handle->m_ownerRunLoop);

    MutexLocker locker(m_streamsMutex);

    handle->m_isDisconnected = true;

    const size_t pos = m_streams.find(handle);
    if (pos == WTF::notFound) {
        LOG_ERROR("SocketStreamHandle: trying to remove unknown handle %p (already deleted?)", handle.get());
        return;
    }
    m_streams.remove(pos);

    curl_easy_cleanup(handle->m_handle);
    handle->m_handle = 0;

    handle->m_socket = -1;
    handle->m_state = SocketStreamHandle::Closed;

    // [Bug 88306]
    // This notification doesn't seem to be needed.
    // wakeUpSocketNotifierThread();
}

void SocketStreamHandleManager::enableReadyWriteNotification(PassRefPtr<SocketStreamHandle> prpHandle)
{
    RefPtr<SocketStreamHandle> handle = prpHandle;

    ASSERT(RunLoop::current() == handle->m_ownerRunLoop);

    MutexLocker locker(m_streamsMutex);

    if (handle->m_hasPendingDataForSending)
        return;

    handle->m_hasPendingDataForSending = true;
    wakeUpSocketNotifierThread();
}

void SocketStreamHandleManager::enableReadyReadNotification(PassRefPtr<SocketStreamHandle> prpHandle)
{
    RefPtr<SocketStreamHandle> handle = prpHandle;

    ASSERT(RunLoop::current() == handle->m_ownerRunLoop);

    MutexLocker locker(m_streamsMutex);

    if (!handle->m_hasPendingReadOrDisconnect)
        return;

    handle->m_hasPendingReadOrDisconnect = false;
    wakeUpSocketNotifierThread();
}

void SocketStreamHandleManager::socketNotifierThread(void* context)
{
    SocketStreamHandleManager* thiz = reinterpret_cast<SocketStreamHandleManager*>(context);

    while (true) {
        fd_set infd, outfd;
        FD_ZERO(&infd);
        FD_ZERO(&outfd);

        FD_SET(thiz->m_readPipeDescriptor, &infd);
        int maxfd = thiz->m_readPipeDescriptor;
        int readfds = 0;
        int writefds = 0;

        thiz->m_streamsMutex.lock();
        for (StreamHandleVector::const_iterator it = thiz->m_streams.begin(); it != thiz->m_streams.end(); ++it) {
            RefPtr<SocketStreamHandle> handle = *it;

            if (handle->isDisconnected())
                continue;

            if (!handle->m_hasPendingReadOrDisconnect) {
                FD_SET(handle->m_socket, &infd);
                ++readfds;
            }
            if (handle->m_hasPendingDataForSending) {
                FD_SET(handle->m_socket, &outfd);
                ++writefds;
            }

            maxfd = std::max(maxfd, static_cast<int>(handle->m_socket));
        }
        thiz->m_streamsMutex.unlock();

        DEBUG_PRINTF("SocketStreamHandleManager::socketNotifierThread: waiting on select() for %d read descriptors and %d write descriptors...\n", readfds, writefds);
        int res = select(maxfd+1, &infd, &outfd, 0, 0);

        if (res > 0) {
            if (FD_ISSET(thiz->m_readPipeDescriptor, &infd)) {
                DEBUG_PRINTF("SocketStreamHandleManager::socketNotifierThread: got wake-up notification!\n");
                // This pipe is just used for receiving wake-up notifications,
                // we don't really care about what we're reading from it
                unsigned char buf[42];
                ::read(thiz->m_readPipeDescriptor, buf, 42);
            }

            thiz->m_streamsMutex.lock();
            for (StreamHandleVector::const_iterator it = thiz->m_streams.begin(); it != thiz->m_streams.end(); ++it) {
                RefPtr<SocketStreamHandle> handle = *it;
                ASSERT(handle->m_socket != -1 || handle->m_socket != 0x29292929);

                // [Bug 88306]
                // A workaround to avoid checking a handle which occurs the hangup.
                // If the handle is deleted, the check for the handle is skipped.
                if (handle->isDisconnected())
                    continue;

                if (FD_ISSET(handle->m_socket, &infd)) {
                    // check for remote disconnect
                    char buf = 0;
                    ssize_t bytesAvailable = recv(handle->m_socket, &buf , sizeof(buf), MSG_PEEK);
                    if (bytesAvailable < 0) {
                        LOG_ERROR("[%d] Error while querying number of bytes available on socket", handle->m_socket);
                        continue;
                    }

                    handle->m_hasPendingReadOrDisconnect = true;

                    if (!bytesAvailable) {
                        DEBUG_PRINTF("[%d] disconnected (remote)\n", handle->m_socket);
                        handle->m_isDisconnected = true;
                        thiz->dispatchRemoteDisconnected(handle);
                        continue;
                    }

                    DEBUG_PRINTF("[%d] %ld bytes available for reading\n", handle->m_socket, (size_t)bytesAvailable);
                    thiz->dispatchReadyRead(handle);
                }
                if (FD_ISSET(handle->m_socket, &outfd)) {
                    DEBUG_PRINTF("[%d] Ready to send more data.\n", handle->m_socket);
                    thiz->dispatchReadyWrite(handle);
                    handle->m_hasPendingDataForSending = false;
                }
            }
            thiz->m_streamsMutex.unlock();
        } else if (!res) {
            // Spurious wake-up (should be rare, but not impossible)
            DEBUG_PRINTF("SocketStreamHandleManager::socketNotifierThread: select() returned without any descriptor set\n");
        } else if (res == -1) {
            // If errno is EBADF we have hit a stale socket, which is a possible situation
            // due to a race condition between the call to m_streamsMutex.unlock() and the call to select()
            // Don't report such errors, it's a known caveat, and we can recover from the error at the next loop iteration
            if (errno != EBADF)
                LOG_ERROR("SocketStreamHandleManager::socketNotifierThread: select() error: %d %s", res, strerror(errno));
        }
    }
}

void SocketStreamHandleManager::wakeUpSocketNotifierThread()
{
    DEBUG_PRINTF("SocketStreamHandleManager::wakeUpSocketNotifierThread()\n");
    ::write(m_writePipeDescriptor, "W", 1);
}

void SocketStreamHandleManager::dispatchReadyRead(PassRefPtr<SocketStreamHandle> prpHandle)
{
    prpHandle->m_ownerRunLoop->dispatch(bind(&SocketStreamHandle::readyRead, prpHandle));
}

void SocketStreamHandleManager::dispatchReadyWrite(PassRefPtr<SocketStreamHandle> prpHandle)
{
    prpHandle->m_ownerRunLoop->dispatch(bind(&SocketStreamHandle::readyWrite, prpHandle));
}

void SocketStreamHandleManager::dispatchRemoteDisconnected(PassRefPtr<SocketStreamHandle> prpHandle)
{
    DEBUG_PRINTF(">>> SocketStreamHandle::%s()\n", __FUNCTION__);
    ASSERT(prpHandle);

    prpHandle->m_ownerRunLoop->dispatch(bind(&SocketStreamHandle::remoteDisconnected, prpHandle));
}

SocketStreamHandleManager::SocketStreamHandleManager()
    : m_socketNotifierThreadId(0)
    , m_readPipeDescriptor(-1)
    , m_writePipeDescriptor(-1)
{
    setupSocketNotifierPipe();

    if (m_readPipeDescriptor != -1 && m_writePipeDescriptor != -1) {
        // Kick off the select() thread
        m_socketNotifierThreadId = createThread(&SocketStreamHandleManager::socketNotifierThread, this, "SocketStreamHandleThread");
        if (!m_socketNotifierThreadId)
            LOG_ERROR("SocketStreamHandleManager: couldn't create socket notifier thread!");
    }
}

void SocketStreamHandleManager::setupSocketNotifierPipe()
{
    int fds[2];
    if (pipe(fds) == -1) {
        LOG_ERROR("SocketStreamHandleManager: couldn't set up socket notifier pipe! (%d)", errno);
        return;
    }

#if defined(F_SETFD) && defined(FD_CLOEXEC)
    fcntl(fds[0], F_SETFD, FD_CLOEXEC);
    fcntl(fds[1], F_SETFD, FD_CLOEXEC);
#endif

    m_readPipeDescriptor = fds[0];
    m_writePipeDescriptor = fds[1];
}

} // namespace WebCore

